package com.pengheng.sink;

import com.pengheng.bean.WaterSensor;
import com.pengheng.transformation.aggregation.WaterSensorMapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * 自定义sink
 * 自定义难实现状态一致性，没有合适的方案的时候使用
 * 推荐使用Flink官方的外部连接器
 */
public class ThemeSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("192.168.10.100", 7777)
                .map(new WaterSensorMapFunction());
        sensorDS.addSink(new SinkFunction<WaterSensor>() {
            @Override
            public void invoke(WaterSensor value, Context context) throws Exception {
                System.out.println("自定义sink接收到的数据：" + value);
                SinkFunction.super.invoke(value, context);
            }
        });
        env.execute();
    }
}
